package com.caul.demo.hadoop.mapreduce.join;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by BlueDream on 2018-04-09.
 */
public class JoinReducer extends Reducer<Text, JoinBean, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<JoinBean> values, Context context) throws IOException, InterruptedException {
        JoinBean profile = null;
        List<JoinBean> orderList = new ArrayList<>();
        for (JoinBean bean : values) {
            if (bean.getType() == JoinBean.FILE_ORDER) {
                orderList.add(bean.clone());
            } else {
                profile = bean.clone();
            }
        }
        //输出
        for (JoinBean bean : orderList) {
            context.write(key, new Text(profile.getUsname() + " " +
                    profile.getAge() + " " +
                    profile.getPartner() + " " +
                    bean.getOrderid()));
        }
    }
}
